package com.raylu.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.raylu.bean.{PageAction, PageDisplay, PageErrorInfo, PageLog, PageStart}
import com.raylu.util.{MyKafkaUtil, OffsetManagerUtil, PropertiesUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.lang
import java.util.Properties
import scala.collection.mutable

/**
 *
 * Description:
 *
 * Create by lucienoz on 2021/11/27.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
object BaseLogApp {

	//1  消费kafka主题
	//2 转换结构
	//3  分流  ：
	//     启动主题 ， DWD_START_LOG  对应pageLog
	//     页面访问主题  DWD_PAGE_LOG
	//     页面动作主题 DWD_PAGE_ACTION
	//     页面曝光主题  DWD_PAGE_DISPLAY
	//     错误主题    DWD_ERROR_INFO
	def main(args : Array[String]) : Unit = {
		val sparkConf : SparkConf = new SparkConf ().setAppName ( "base_log_app" ).setMaster ( "local[*]" )
		val ssc = new StreamingContext ( sparkConf, Seconds ( 5 ) )

		val properties : Properties = PropertiesUtil.load ( "config.properties" )
		val topic = properties.getProperty("kafka.log.topic")
		val groupId = "base_log_group"
		val offset : Map[TopicPartition, Long] = OffsetManagerUtil.getOffset ( topic, groupId )

		var KafkaDStream : InputDStream[ConsumerRecord[String, String]] = null
		if (offset.isEmpty && offset == null) KafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( topic, ssc, groupId )
		else KafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( topic, ssc, groupId, offset )

		var offsetRanges : Array[OffsetRange] = null
		val KafkaDStream2 : DStream[ConsumerRecord[String, String]] = KafkaDStream.transform { RDD =>
			offsetRanges = RDD.asInstanceOf[HasOffsetRanges].offsetRanges
			RDD
		}

		KafkaDStream2.map { consumerRecord : ConsumerRecord[String, String] =>
			consumerRecord.value ()
			JSON.parseObject ( consumerRecord.value () )
		}.foreachRDD {
			RDD =>
				RDD.foreachPartition { jsonObjIter =>
					jsonObjIter.foreach {
						jsonObj =>
							val topicRecord : mutable.Set[String] = mutable.Set[String]()
							//common field
							val commonJSONObj : JSONObject = jsonObj.getJSONObject ( "common" )
							val ar : String = commonJSONObj.getString ( "ar" )
							val ba : String = commonJSONObj.getString ( "ba" )
							val ch : String = commonJSONObj.getString ( "ch" )
							val isNew : String = commonJSONObj.getString ( "is_new" )
							val md : String = commonJSONObj.getString ( "md" )
							val mid : String = commonJSONObj.getString ( "mid" )
							val os : String = commonJSONObj.getString ( "os" )
							val uid : String = commonJSONObj.getString ( "uid" )
							val vc : String = commonJSONObj.getString ( "vc" )
							val ts : lang.Long = jsonObj.getLong ( "ts" )
							//page field
							val pageJSONObj : JSONObject = jsonObj.getJSONObject ( "page" )
							var duringTime : String = null
							var lastPageId : String = null
							var pageId : String = null
							if (pageJSONObj != null && pageJSONObj.size () > 0) {
								topicRecord.add ( "DWD_PAGE_LOG" )
								duringTime = pageJSONObj.getString ( "during_time" )
								lastPageId = pageJSONObj.getString ( "last_page_id" )
								pageId = pageJSONObj.getString ( "page_id" )
							}

							//display filed
							val displaysJSONObj : JSONArray = jsonObj.getJSONArray ( "displays" )
							val displayType : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val item : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val itemType : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val order : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val posId : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							if (displaysJSONObj != null && displaysJSONObj.size () > 0) {
								topicRecord.add ( "DWD_PAGE_DISPLAY" )
								for (i <- 0 until displaysJSONObj.size ()) {
									val displayJSONObj : JSONObject = displaysJSONObj.getJSONObject ( i )
									displayType.append(displayJSONObj.getString("display_type"))
									item.append(displayJSONObj.getString("item"))
									itemType.append(displayJSONObj.getString("item_type"))
									order.append(displayJSONObj.getString("order"))
									posId.append(displayJSONObj.getString("pos_id"))

								}
							}
							//action
							val actionsJSONObj : JSONArray = jsonObj.getJSONArray ( "actions" )
							val actionId : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val actItem : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val actItemType : mutable.ListBuffer[String] = mutable.ListBuffer[String]()
							val actTs : mutable.ListBuffer[Long] = mutable.ListBuffer[Long]()
							if (actionsJSONObj != null && actionsJSONObj.size () > 0) {
								topicRecord.add ( "DWD_PAGE_ACTION" )
								for (i <- 0 until actionsJSONObj.size ()) {
									val actionJSONObj : JSONObject = actionsJSONObj.getJSONObject ( i )
									actionId.append(actionJSONObj.getString("action_id"))
									actItem.append(actionJSONObj.getString("item"))
									actItemType.append(actionJSONObj.getString("item_type"))
									actTs.append(actionJSONObj.getLong("ts"))
								}
							}

							//err

							val errJSONObj: JSONObject = jsonObj.getJSONObject("err")
							var errorCode:String = null
							var msg:String = null
							if (errJSONObj!=null&&errJSONObj.size()>0) {
								topicRecord.add ( "DWD_ERROR_INFO" )
								errorCode=errJSONObj.getString("error_code")
								msg=errJSONObj.getString("msg")
							}

							//start field
							val startJSONObj: JSONObject = jsonObj.getJSONObject("start")
							var entry:String = null
							var loadingTime:String = null
							var openAdId:String = null
							var openAdMs:String = null
							var openAdSkipMs:String = null
							if (startJSONObj!=null&&startJSONObj.size()>0) {
								topicRecord.add ( "DWD_START_LOG" )
								entry=startJSONObj.getString("entry")
								loadingTime=startJSONObj.getString("loading_time")
								openAdId=startJSONObj.getString("open_ad_id")
								openAdMs=startJSONObj.getString("open_ad_ms")
								openAdSkipMs=startJSONObj.getString("open_ad_skip_ms")
							}

							if (topicRecord.contains("DWD_START_LOG")) {
								val pageStart : PageStart = PageStart ( ar, ba, ch, isNew, md, mid, os, uid, vc, ts, entry, loadingTime, openAdId, openAdMs, openAdSkipMs )
								MyKafkaUtil.send("DWD_START_LOG",JSON.toJSONString(pageStart,new SerializeConfig(true)))
							}
							if (topicRecord.contains("DWD_ERROR_INFO")) {
								val pageErrorInfo : PageErrorInfo = PageErrorInfo ( ar, ba, ch, isNew, md, mid, os, uid, vc, ts, errorCode, msg )
								MyKafkaUtil.send("DWD_ERROR_INFO",JSON.toJSONString(pageErrorInfo,new SerializeConfig(true)))

							}
							if (topicRecord.contains("DWD_PAGE_ACTION")) {
								for(i <- 0 until actionId.size){
									val pageAction : PageAction = PageAction ( ar, ba, ch, isNew, md, mid, os, uid, vc, ts, actionId ( i ), actItem ( i ), actItemType ( i ), actTs ( i ) )
									MyKafkaUtil.send("DWD_PAGE_ACTION",JSON.toJSONString(pageAction,new SerializeConfig(true)))

								}
							}
							
							
							if (topicRecord.contains("DWD_PAGE_DISPLAY")) {
								for(i <- 0 until displayType.size){
									val pageDisplay : PageDisplay = PageDisplay ( ar, ba, ch, isNew, md, mid, os, uid, vc, ts, displayType ( i ), item ( i ), itemType ( i ), order ( i ), posId ( i ) )
									MyKafkaUtil.send("DWD_PAGE_DISPLAY",JSON.toJSONString(pageDisplay,new SerializeConfig(true)))
								}
							}

							if (topicRecord.contains("DWD_PAGE_LOG")) {
								val pageLog : PageLog = PageLog ( ar, ba, ch, isNew, md, mid, os, uid, vc, ts, duringTime, lastPageId, pageId )
								MyKafkaUtil.send("DWD_PAGE_LOG",JSON.toJSONString(pageLog,new SerializeConfig(true)))
							}
							

					}
					MyKafkaUtil.flush()

				}
				OffsetManagerUtil.saveOffset(topic, groupId, offsetRanges)
		}

		ssc.start ()
		ssc.awaitTermination ()


	}


}
